package com.kyu.LogDiff;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

public class Map extends Mapper<LongWritable,Text,Text,Bean> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        Bean bean = new Bean();
        String line = value.toString();
        if (StringUtils.isNoneBlank(line)) {
            String fleids[] = line.split("\\001");
            bean.setTemper(Double.parseDouble(fleids[9]));                          // 取温度、气压，计算温度露点差
            bean.setPress(Double.parseDouble(fleids[23]));
            bean.setTemper_diffdew(new BigDecimal(fleids[9]).subtract(new BigDecimal(fleids[10])).doubleValue());
            SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd HHmm");   // 生成3小时、6小时、24小时差
            Calendar calendar = Calendar.getInstance();
            try {
                Date d = sdf.parse(fleids[0]+" "+fleids[1]);
                calendar.setTime(d);
            } catch (ParseException e) { e.printStackTrace(); }
            bean.setOffset(0);
            String datetime = sdf.format(calendar.getTime());
            System.out.println(datetime);
            context.write(new Text(datetime),bean);                      // 输出当前数据

            bean.setOffset(3);
            calendar.add(Calendar.HOUR_OF_DAY,3);              // 3小时
            datetime = sdf.format(calendar.getTime());
            System.out.println(datetime);
            context.write(new Text(datetime),bean);                      // 输出+3小时数据

            bean.setOffset(6);
            calendar.add(Calendar.HOUR_OF_DAY,3);              // 6小时
            datetime = sdf.format(calendar.getTime());
            context.write(new Text(datetime),bean);                      // 输出+6小时数据

            bean.setOffset(24);
            calendar.add(Calendar.HOUR_OF_DAY,18);             // 24小时
            datetime = sdf.format(calendar.getTime());
            context.write(new Text(datetime),bean);                      // 输出+24小时数据
        }
    }
}
